package query;

import java.io.IOException;

import libMapReduce.MultiTableInputCollection;
import libMapReduce.TableMapReduceUtil;



import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
//import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;

public class FreqCounter1 {

    static class Mapper1 extends TableMapper<ImmutableBytesWritable, IntWritable> {

        private int numRecords = 0;
        private static final IntWritable one = new IntWritable(1);

        @Override
        public void map(ImmutableBytesWritable row, Result values, Context context) throws IOException {
            // extract userKey from the compositeKey (userId + counter)
//            ImmutableBytesWritable userKey = new ImmutableBytesWritable(row.get(), 0, Bytes.SIZEOF_INT);
            ImmutableBytesWritable userKey = new ImmutableBytesWritable(row.get(), 0, 1);
            try {
//            	System.out.println("key:"+row.toString()+"\tvalues:"+values.toString());
//            	System.out.println();
                context.write(userKey, one);
            } catch (InterruptedException e) {
                throw new IOException(e);
            }
//            System.out.println(userKey);
            numRecords++;
            if ((numRecords % 100) == 0) {
                context.setStatus("mapper processed " + numRecords + " records so far");
            }
        }
    }

    public static class Reducer1 extends TableReducer<ImmutableBytesWritable, IntWritable, ImmutableBytesWritable> {

        public void reduce(ImmutableBytesWritable key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }

            Put put = new Put(key.get());
            put.add(Bytes.toBytes("cf1"), Bytes.toBytes("total"), Bytes.toBytes(sum));
            System.out.println(String.format("stats :   key : %s  count : %d", key.get().toString(), sum));
            context.write(key, put);
        }
    }
    
    public static void main(String[] args) throws Exception {
    	Configuration configuration = HBaseConfiguration.create();//new HBaseConfiguration();
        
        

//        String columns = "details"; // comma seperated
//        scan.addColumns(columns);
//        scan.addColumns(columns);
//        configuration = HBaseConfiguration.create();
		configuration.set("hbase.master", "192.9.200.169:44696");
		configuration.set("hbase.zookeeper.quorum","192.9.200.169");  
		configuration.set("hbase.zookeeper.property.clientPort", "2181");    
		configuration.set("hbase.HServerAddress", "192.9.200.169:44696");
		Job job = new Job(configuration, "Hbase_FreqCounter1");
        job.setJarByClass(FreqCounter1.class);
		
//        scan.setFilter(new FirstKeyOnlyFilter());
		Scan scan1 = new Scan();
		scan1.setStartRow(null);
		scan1.setStopRow(null);
		String family="cf1";
//		String family2="cf2";
		scan1.addFamily(family.getBytes());
//		scan1.addFamily(family2.getBytes());
		
		Scan scan2 = new Scan();
		scan2.setStartRow(null);
		scan2.setStopRow(null);
		String cf1="cf1";
		String c1="c1";
		String cf2="cf2";
		String c2="c2";
//		scan2.addColumn(cf1.getBytes(), c1.getBytes());
		scan2.addColumn(cf2.getBytes(), c2.getBytes());
//		scan1.addFamily(family22.getBytes());
//		scan2.addColumn(family2.getBytes(), column2.getBytes());
		MultiTableInputCollection mit = new MultiTableInputCollection();
		String table1 = "TH_1M";
		String table2 = "DH_1M";
		
		mit.Add(table1, scan1);
		mit.Add(table2, scan2);
		
		
		
//		TableMapReduceUtil.initTableMapperJob(table1.getBytes(), scan1, Mapper1.class, ImmutableBytesWritable.class, IntWritable.class, job);
//		TableMapReduceUtil.initTableMapperJob(table1.getBytes(), scan2, Mapper1.class, ImmutableBytesWritable.class, IntWritable.class, job);
		
		TableMapReduceUtil.initTableMapperJob(mit, Mapper1.class, ImmutableBytesWritable.class,
				IntWritable.class, job);
		
//        TableMapReduceUtil.initTableMapperJob("test", scan, Mapper1.class, ImmutableBytesWritable.class,
//                IntWritable.class, job);
//        TableMapReduceUtil.initTableMapperJob("mytable", scan, Mapper1.class, ImmutableBytesWritable.class,
//                IntWritable.class, job);

        TableMapReduceUtil.initTableReducerJob("result", Reducer1.class, job);
        long startTime = System.currentTimeMillis();
        job.waitForCompletion(true);
        System.out.println("Running time:"+(System.currentTimeMillis()-startTime)/1000.00 + "seconds");
    }
}